package com.dongdongshop.mq;


import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.dongdongshop.pojo.salePoints.SalePoints;
import com.dongdongshop.service.SalePointServiceI;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;

import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.List;

@Component
@RocketMQMessageListener(consumerGroup = "my-order",topic = "countAndOrder")
public class PointsConsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Reference
    private SalePointServiceI salePointServiceI;
    @Resource
    private RedisTemplate redisTemplate;

    @Override
    public void onMessage(String s) {
        //普通消费方法 要发送的东西
        System.out.println("要发送的东西"+s);
    }

    /**
     * 自定义消息消费重试策略（手动接管ACK确认机制）
     * CONSUME_SUCCESS、RECONSUME_LATER
     * 注意：保留原有消费者监听RocketMQListener
     */
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt message = list.get(0); //消息在list中
                String messageBody =null;
                try {

                   System.out.println("消息在list中"+message);
                   //获取message里的内容 message.getBody();
                  messageBody = new String(message.getBody(), "utf-8");
                   //System.out.println("内容"+messageBody);
                   List<String> point = JSON.parseArray(messageBody, String.class);
                   //根据id查询 存在修改 不存在增加
                   String userId = point.get(1);
                   BigDecimal price = new BigDecimal(point.get(2));
                   Double aDouble = Double.valueOf(point.get(2)).intValue()*0.1;
                   int points=new Double(aDouble).intValue();
                   SalePoints salePoints=salePointServiceI.selectByUserId(userId);
                   //如果用户存在
                   if (salePoints != null){
                       //消费金额
                       salePoints.setSalePrice(salePoints.getSalePrice().add(price));
                       //添加积分
                       salePoints.setSalePoit(salePoints.getSalePoit()+points);
                       //修改积分
                       salePointServiceI.updatePoints(salePoints);
                   }else {
                       SalePoints salePoints1=new SalePoints();
                       salePoints1.setUserName(point.get(0));
                       salePoints1.setUserId(point.get(1));
                       salePoints1.setSalePrice(price);
                       salePoints1.setSalePoit(points);
                       salePoints1.setSaleGrade("青铜");
                       //增加
                       salePointServiceI.insertSalePoints(salePoints1);
                   }
                   System.out.println("ACK消息消费成功"+messageBody);
               }catch (Exception e){
                   e.printStackTrace();
                   //消息id
                    String msgId = message.getMsgId();
                    //重新消费的次数（重试次数）
                    int times = message.getReconsumeTimes();
                    if (times >=2){
                        //则会跳消息可以存储到db或者log日志中 或者其他处理方式 然后重试
                        redisTemplate.opsForValue().set(msgId,messageBody);
                        System.out.println("失败次数过多，进入死信队列");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }else {
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }



}
